package com.lichao.cdc;

/**
 * @author: li chao
 * @Desc:
 * @create: 2024-07-11 09:52
 **/

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkCDC_SQL {

    public static void main(String[] args) {

        //1.获取Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.使用FlinkCDC SQL方式建表
        tableEnv.executeSql("" +
                "create table t1(\n" +
                " id string primary key NOT ENFORCED,\n" +
                " name string " +
                ") with (\n" +
                " 'connector' = 'mysql-cdc',\n " +
                " 'hostname' = 'localhost',\n " +
                " 'port' = '3306',\n " +
                " 'username' = 'root',\n " +
                " 'password' = '123456',\n " +
                " 'database-name' = 'learning',\n " +
                " 'table-name' = 't1'\n " +
                ")");


        //3.查询并打印
        Table table = tableEnv.sqlQuery("select * from t1");
        table.execute().print();
    }

}
